package io.kiki.sba.registry.server.clientInterface.circuit.breaker;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import io.kiki.sba.registry.cache.CacheCleaner;
import io.kiki.sba.registry.common.model.store.CircuitBreakerStatistic;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchCircuitBreakerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;
import java.util.concurrent.TimeUnit;


public class DefaultCircuitBreakerService implements CircuitBreakerService {
    public static final Logger logger = LoggerFactory.getLogger(DefaultCircuitBreakerService.class);

    private final Cache<String, CircuitBreakerStatistic> cache = CacheBuilder.newBuilder().maximumSize(2000L).expireAfterAccess(10, TimeUnit.MINUTES).build();
    @Autowired
    protected FetchCircuitBreakerService fetchCircuitBreakerService;
    @Autowired
    private Config config;

    public DefaultCircuitBreakerService() {
        CacheCleaner.autoClean(cache, 1000 * 60 * 10);
    }

    /**
     * check if statistic should be circuit break
     */
    @Override
    public boolean pushCircuitBreaker(CircuitBreakerStatistic circuitBreakerStatistic, boolean hasPushed) {
        if (circuitBreakerStatistic == null) {
            return false;
        }

        // not circuit break on sub.register
        if (!hasPushed) {
            return false;
        }

        if (fetchCircuitBreakerService.isSwitchOpen()) {
            return addressCircuitBreak(circuitBreakerStatistic);
        }
        int silenceMillis = config.getPushCircuitBreakerSilenceMillis();
        return circuitBreakerStatistic.circuitBreak(config.getPushCircuitBreakerThreshold(), silenceMillis);
    }

    protected boolean addressCircuitBreak(CircuitBreakerStatistic circuitBreakerStatistic) {
        if (fetchCircuitBreakerService.getStopPushCircuitBreaker().contains(circuitBreakerStatistic.getIp())) {
            logger.info("[ArtificialCircuitBreaker]push check circuit break, statistic:{}", circuitBreakerStatistic);
            return true;
        }
        int silenceMillis = config.getPushCircuitBreakerSilenceMillis();
        CircuitBreakerStatistic circuitBreakerStatistic1 = cache.getIfPresent(circuitBreakerStatistic.getAddress());
        if (circuitBreakerStatistic1 != null) {
            boolean addressCircuitBreak = circuitBreakerStatistic1.circuitBreak(config.getPushAddressCircuitBreakerThreshold(), silenceMillis);
            boolean subCircuitBreak = circuitBreakerStatistic.circuitBreak(config.getPushCircuitBreakerThreshold(), silenceMillis);
            logger.info("[addressCircuitBreak]addressCircuitBreak: {}, addressStatistic:{}, subCircuitBreak:{}, subStatistic:{}", addressCircuitBreak, circuitBreakerStatistic1, subCircuitBreak, circuitBreakerStatistic);
            return addressCircuitBreak || subCircuitBreak;
        }

        return circuitBreakerStatistic.circuitBreak(config.getPushCircuitBreakerThreshold(), silenceMillis);
    }

    /**
     * statistic when push fail
     */
    public boolean onPushFail(Map<String, Long> versions, Subscriber subscriber) {

        if (!subscriber.hasPushed()) {
            // push fail on register, not circuit breaker;
            return true;
        }
        String ip = subscriber.getClientUrl().getIp();
        String address = subscriber.getClientUrl().buildAddressString();
        if (!subscriber.onPushFail(versions)) {
            logger.info("PushN, failed to do onPushFail, {}, {}", subscriber.getDataInfoId(), address);
            return false;
        }

        try {
            CircuitBreakerStatistic circuitBreakerStatistic = cache.get(address, () -> new CircuitBreakerStatistic(subscriber.getGroup(), ip, address));
            circuitBreakerStatistic.fail();
            logger.info("PushN, dataInfoId={}, inc circuit statistic={}", subscriber.getDataInfoId(), circuitBreakerStatistic);
            return true;
        } catch (Throwable e) {
            logger.error("[onPushFail]get from circuitBreakerAddress error.", e);
            return false;
        }
    }

    /**
     * statistic when push success
     */
    public boolean onPushSuccess(Map<String, Long> versions, Map<String, Integer> pushNums, Subscriber subscriber) {
        String address = subscriber.getClientUrl().buildAddressString();

        CircuitBreakerStatistic circuitBreakerStatistic = cache.getIfPresent(address);
        if (circuitBreakerStatistic != null) {
            int threshold = config.getPushConsecutiveSuccess();
            circuitBreakerStatistic.success(threshold);
            if (circuitBreakerStatistic.getConsecutiveSuccess() >= threshold) {
                cache.invalidate(address);
                logger.info("PushY, invalidate circuit statistic: {}", circuitBreakerStatistic);
            }
        }

        if (subscriber.checkAndUpdatePushContext(versions, pushNums)) {
            logger.info("PushY, checkAndUpdateCtx onPushSuccess, {}, {}, versions={}", subscriber.getDataInfoId(), address, versions);
            return true;
        } else {
            logger.info("PushN, failed to checkAndUpdateCtx onPushSuccess, {}, {}", subscriber.getDataInfoId(), address);
            return false;
        }
    }


}
